package com.ruoyi.chat.sse;

import com.ruoyi.common.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 流式处理客户端
 */
@Component
public class SseClient {

    private static final Logger log = LoggerFactory.getLogger(SseClient.class);

    private static final Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    /**
     * 创建连接
     *
     * @param uid 连接id
     * @return 连接实例
     */
    public SseEmitter createSse(String uid) {
        //默认30秒超时,设置为0L则永不超时,此处设置10分钟
        SseEmitter sseEmitter = new SseEmitter(600 * 1000L);
        //完成后回调
        sseEmitter.onCompletion(() -> {
            log.info("[{}]结束连接...................", uid);
            sseEmitterMap.remove(uid);
        });
        //超时回调
        sseEmitter.onTimeout(() -> {
            log.info("[{}]连接超时...................", uid);
            sseEmitter.complete();
        });
        //异常回调
        sseEmitter.onError(
                throwable -> {
                    try {
                        log.info("[{}]连接异常,{}", uid, throwable.toString());
                        sseEmitter.send(SseEmitter.event()
                                .id(uid)
                                .name("发生异常！")
                                .data("发生异常请重试！")
                                .reconnectTime(3000));
                        sseEmitterMap.put(uid, sseEmitter);
                    } catch (IOException e) {
                        log.error("sseEmitter发送异常,{}", e.getMessage());
                        sseEmitter.completeWithError(e);
                    }
                }
        );
        sseEmitterMap.put(uid, sseEmitter);
        log.info("[{}]创建sse连接成功！", uid);
        return sseEmitter;
    }

    @Async(value = "threadPoolTaskExecutor")
    public static void send(SseEmitter sseEmitter, String message, String apiKey) {
        if (StringUtils.isBlank(message)) {
            log.info("answer为空，uid:[{}]", apiKey);
            return;
        }
        try {
            sseEmitter.send(SseEmitter.event().name("message").reconnectTime(60*1000L).data(message));
            log.info("用户{},推送成功:{}", apiKey, message);
        }catch (Exception e) {
            log.info("用户{},推送异常:{}", apiKey, e.getMessage());
            sseEmitter.complete();
        }
    }

    /**
     * 手动断开
     * @param uid uid
     */
    public void closeSse(String uid){
        if (sseEmitterMap.containsKey(uid)) {
            SseEmitter sseEmitter = sseEmitterMap.get(uid);
            sseEmitter.complete();
            sseEmitterMap.remove(uid);
        }else {
            log.info("用户{} 连接已关闭",uid);
        }

    }


}